package xin.marcher.module.inventory.mq.consumer.listener;

import com.google.common.cache.Cache;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import xin.marcher.module.common.cache.MessageCache;
import xin.marcher.module.common.constants.inventory.TairInventoryConstant;
import xin.marcher.module.common.enums.inventory.BucketStatusEnum;
import xin.marcher.module.common.utils.DateFormatUtil;
import xin.marcher.module.common.utils.JsonUtil;
import xin.marcher.module.inventory.cache.InventoryBucketCache;
import xin.marcher.module.inventory.domain.bo.BucketLocalCache;

import javax.annotation.Resource;
import java.util.Date;
import java.util.List;
import java.util.Objects;

/**
 * 处理 库存扣减明细的记录消息
 */
@Slf4j
@Component
public class CacheRefreshListener implements MessageListenerConcurrently {

    /**
     * 本地缓存
     */
    @Resource
    private Cache cache;

    @Resource
    private InventoryBucketCache inventoryBucketCache;

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        try {
            for (MessageExt messageExt : list) {
                String msg = new String(messageExt.getBody());
                log.info("刷新本地缓存，消息内容：{},消费时间:{}", msg, DateFormatUtil.formatDateTime(new Date()));

                MessageCache messageCache = JsonUtil.json2Object(msg, MessageCache.class);

                BucketLocalCache bucketLocalCache = JsonUtil.json2Object(messageCache.getCacheJSON(), BucketLocalCache.class);
                synchronized (messageCache.getCacheKey().intern()) {
                    String bucketLocalKey = TairInventoryConstant.SELLER_BUCKET_PREFIX + bucketLocalCache.getSellerId() + bucketLocalCache.getSkuId();
                    // 分桶的真实下线，由实际的下线分桶远程元数据变更为准
                    BucketLocalCache bucketCache = inventoryBucketCache.getTairBucketCache(bucketLocalKey);

                    if (Objects.isNull(bucketCache)) {
                        cache.put(messageCache.getCacheKey(), JsonUtil.json2Object(messageCache.getCacheJSON(), BucketLocalCache.class));
//                        log.error("更新本地缓存，本次更新内容{},更新时间{}", messageCache.getCacheJSON(), DateFormatUtil.formatDateTime(new Date()));
                    } else {
                        // 进行diff数据集处理
                        if (bucketLocalCache.getOperationType().equals(BucketStatusEnum.AVAILABLE_STATUS.getCode())) {
                            inventoryBucketCache.diffCacheOnline(bucketCache, bucketLocalCache);
                        } else if (bucketLocalCache.getOperationType().equals(BucketStatusEnum.OFFLINE_STATUS.getCode())) {
                            inventoryBucketCache.diffCacheOffline(bucketCache, bucketLocalCache);
                        }
                        cache.put(messageCache.getCacheKey(), bucketCache);
//                        log.error("更新本地缓存，本次更新内容{},更新时间{}", JSONObject.toJSONString(bucketCache), DateFormatUtil.formatDateTime(new Date()));
                    }
                }
            }
        } catch (Exception e) {
            // 本地缓存只有参数转换会出错，这种错误重试也没什么作用
            log.error("consume error, 刷新本地缓存失败", e);
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }


}
